package com.xhg.mqtt.core;

import com.google.protobuf.ByteString;
import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.constant.ResultCodeConstants;
import com.xhg.mqtt.dto.EventCodeEnum;
import com.xhg.mqtt.dto.MqttConfigInfoDTO;
import com.xhg.mqtt.dto.MqttMessagePb;
import com.xhg.mqtt.dto.OnlineStatusPb;
import com.xhg.mqtt.job.MessageRepeatQueue;
import com.xhg.mqtt.listener.MessageEvent;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import com.xhg.mqtt.util.CommonUtil;
import com.xhg.mqtt.util.ProtoBufUtil;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * mqtt消息处理
 *
 * @author E.T
 * @version 1.0
 * @Description
 * @date 2018-08-01 16:16
 * @updateby
 * @updatedate
 * @since 1.0
 */
public class MqttMessageHandler {

    private MqttConfigInfoDTO mqttInfo;
    private Map<String, String> map = new HashMap<>();

    public MqttMessageHandler() {
    }

    public MqttMessageHandler(MqttConfigInfoDTO mqttInfo) {
        this.mqttInfo = mqttInfo;
        if (mqttInfo != null) {
            map.put("{province}", mqttInfo.getProvince());
            map.put("{city}", mqttInfo.getCity());
            map.put("{area}", mqttInfo.getArea());
        }
    }

    public boolean handleMessage(EventCodeEnum eventCodeEnum, String deviceId, String version, int cc,int qos,
        ByteString body)
        throws Exception {
        map.put("{clientId}", deviceId);
        map.put("{eventCode}", eventCodeEnum.getCode());
//        String messageId = "123456";
        String messageId = CommonUtil.getUUID("");
//        String pushTopic = "hello";
        String pushTopic = mqttInfo.getPushTopic();
        //构建消息头
        MqttMessagePb.MqttHead.Builder head = ProtoBufUtil
            .builderHead(eventCodeEnum.getCode(), messageId, deviceId, version, cc,
                ResultCodeConstants.SUCCESS, null);
        //组装消息
        byte[] bytes = ProtoBufUtil.builderMessage(head, body);
        String topic = CommonUtil.replaceTopic(pushTopic, map);
        if (MqttSimpleClient.getInstance() !=null && MqttSimpleClient.getInstance().isConnect()) {
            MqttSimpleClient.getInstance().publish(topic,qos, bytes);
            if (CommonUtil.isAck(cc)) {
                MessageRepeatQueue.getInstance().add(messageId,
                    new MessageEvent(eventCodeEnum.getCode(), topic, bytes,
                        System.currentTimeMillis()));
            }
            MqttLoggerFactory.info(
                "设备：" + deviceId + " 发送消息【" + messageId + "】到topic->" + topic + " 指令码:"
                    + eventCodeEnum
                    .getCode() + "指令信息:"
                    + eventCodeEnum.getDesc());
        }else{
            MqttLoggerFactory.error(MsgConstants.MQTT_NOT_CONN);
            return false;
        }
        return true;
    }

    public boolean ackMessage(EventCodeEnum eventCodeEnum, String messageId, String deviceId,
        String version,int qos)
        throws Exception {
        map.put("{clientId}", deviceId);
        map.put("{eventCode}", eventCodeEnum.getCode());
//        String ackTopic = "xhg/hello/ack";
        String ackTopic = mqttInfo.getAckTopic();
        //构建消息头
        MqttMessagePb.MqttHead.Builder head = ProtoBufUtil
            .builderHead(eventCodeEnum.getCode(), messageId, deviceId, version, 0,
                ResultCodeConstants.SUCCESS, null);
        //组装消息
        byte[] bytes = ProtoBufUtil.builderMessage(head, null);
        String topic = CommonUtil.replaceTopic(ackTopic, map);
        if (MqttSimpleClient.getInstance() !=null && MqttSimpleClient.getInstance().isConnect()) {
            MqttSimpleClient.getInstance().publish(topic,qos, bytes);
            MqttLoggerFactory.info(
                "设备：" + deviceId + " 发送ack消息【" + messageId + "】到topic->" + topic + " 指令码:"
                    + eventCodeEnum.getCode() + "指令信息:"
                    + eventCodeEnum.getDesc());
        }else{
            MqttLoggerFactory.error(MsgConstants.MQTT_NOT_CONN);
            return false;
        }
        return true;
    }

    /**
     * 设备上报
     */
    public void login(String deviceId, String version) {
        ScheduledExecutorService service= Executors.newSingleThreadScheduledExecutor();
        service.schedule(() -> {
            try {
                //构建业务消息内容
                OnlineStatusPb.OnlineStatus.Builder online = OnlineStatusPb.OnlineStatus
                    .newBuilder();
                online.setStatus(1);
                online.setStatusDesc("[" + deviceId + "]" + " 设备上线");
                handleMessage(EventCodeEnum.DEVICE_LOGIN, deviceId, version, 1,MqttConstants.QOS,
                    online.build().toByteString());
            } catch (Exception e) {
                MqttLoggerFactory.error(MsgConstants.DEVICE_UP_ERROR, e);
            }
        }, CommonUtil.getRandomTime(), TimeUnit.SECONDS);
        service.shutdown();
    }

}
